package day07

import day07.caseClass.CategoryCountInfo
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD


/**
 * 6.3	需求2：Top10热门品类中每个品类的Top10活跃Session统计
 *
 * 1)	需求描述
 * 对于排名前10的品类，分别获取每个品类点击次数排名前10的sessionId。（注意: 这里我们只关注点击次数，不关心下单和支付次数）
 * 这个就是说，对于top10的品类，每一个都要获取对它点击次数排名前10的sessionId。
 * 这个功能，可以让我们看到，对某个用户群体最感兴趣的品类，各个品类最感兴趣最典型的用户的session的行为。
 *
 * 2)	分析思路
 * 	通过需求1，获取TopN热门品类的id
 * 	将原始数据进行过滤（1.保留热门品类 2.只保留点击操作）
 * 	对session的点击数进行转换 (category~session,1)
 * 	对session的点击数进行统计 (category~session,sum)
 * 	将统计聚合的结果进行转换  (category,(session,sum))
 * 	将转换后的结构按照品类进行分组 (category,Iterator[(session,sum)])
 * 	对分组后的数据降序 取前10
 *
 * 6.3.2	代码实现
 */
object Spark01_TopN_Req2 {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    val actionRDD: RDD[String] = sc.textFile(path = "E:\\学习资料\\user_visit_action.txt")

    // 通过需求1，获取TopN热门品类的id
    val topNList: List[CategoryCountInfo] = TopNUtils.getTopNList(sc)

    // 获取top10中的品类id
    val categoryIdList: List[String] = topNList.map(_.categoryId)

    val categoryIdsBroadcast: Broadcast[List[String]] = sc.broadcast(categoryIdList)

    // 将原始数据进行过滤（1.保留热门品类 2.只保留点击操作）
    val topNClickActionRDD: RDD[String] = actionRDD.filter {
      userVisitAction => {
        val properties: Array[String] = userVisitAction.split("_")
        val clickCategoryId: Long = properties(6).toLong
        val clickProductId: Long = properties(7).toLong
        val topNCategoryIds: List[String] = categoryIdsBroadcast.value
        topNCategoryIds.contains(clickCategoryId.toString) && TopNUtils.clickAction(clickCategoryId, clickProductId)
      }
    }

    // 对session的点击数进行转换 (category~session,1)
    val topNMapRdd: RDD[(String, Int)] = topNClickActionRDD.map {
      userVisitAction => {
        val properties: Array[String] = userVisitAction.split("_")
        val sessionId: String = properties(2)
        val clickCategoryId: String = properties(6)
        (clickCategoryId.concat("~").concat(sessionId), 1)
      }
    }

    // 对session的点击数进行统计 (category~session,sum)
    val reduceRdd: RDD[(String, Int)] = topNMapRdd.reduceByKey(_ + _)

    // 将统计聚合的结果进行转换  (category,(session,sum))
    val mapRdd: RDD[(String, (String, Int))] = reduceRdd.map {
      case (str, count) =>
        val array: Array[String] = str.split("~")
        val categoryId: String = array(0)
        val sessionId: String = array(1)
        (categoryId, (sessionId, count))
    }

    // 将转换后的结构按照品类进行分组 (category,Iterator[(session,sum)])
    val groupRdd: RDD[(String, Iterable[(String, Int)])] = mapRdd.groupByKey

    // 对分组后的数据降序 取前10
    val resRdd: RDD[(String, List[(String, Int)])] = groupRdd.mapValues {
      iterable => iterable.toList.sortWith((t1, t2) => t1._2 > t2._2).take(n = 10)
    }

    resRdd.foreach(println)

    // 关闭连接
    sc.stop()
  }
}
